package org.jetlinks.pro.clickhouse;

import io.netty.util.internal.ThreadLocalRandom;
import lombok.Getter;
import lombok.Setter;
import org.jetlinks.pro.clickhouse.engine.ClickHouseEngineTemplate;
import org.jetlinks.pro.clickhouse.engine.EngineType;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.util.StringUtils;
import org.springframework.util.unit.DataSize;
import org.springframework.web.reactive.function.client.*;
import org.springframework.web.util.UriComponentsBuilder;
import reactor.core.publisher.Mono;

import javax.annotation.Nonnull;
import java.net.URI;
import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

@Getter
@Setter
@ConfigurationProperties(prefix = "clickhouse")
public class ClickHouseProperties {

    private int maxBatchSize = 10000;

    private int maxRetry = 3;

    private Duration retryDuration = Duration.ofSeconds(1);

    private Duration batchDuration = Duration.ofSeconds(2);

    //表引擎模版配置
    private ClickHouseEngineTemplate template = new ClickHouseEngineTemplate();

    private RestfulConnector restful = new RestfulConnector();

    @Getter
    @Setter
    public static class RestfulConnector {

        private String url = "http://127.0.0.1:8123";

        private List<URI> endpoints = new ArrayList<>();

        private String username = "default";

        private String password = "jetlinks";

        private String database;

        public URI selectURI() {
            // TODO: 2021/6/2 更好的负载均衡方式
            return endpoints.get(ThreadLocalRandom.current().nextInt(endpoints.size()));
        }

        public WebClient create() {
            WebClient.Builder builder = WebClient.builder();

            return builder
                .baseUrl(url)
                .filter(new ExchangeFilterFunction() {
                    @Nonnull
                    @Override
                    public Mono<ClientResponse> filter(@Nonnull ClientRequest request,
                                                       @Nonnull ExchangeFunction next) {
                        if (endpoints.size() > 0) {
                            URI target = selectURI();
                            URI uri = UriComponentsBuilder
                                .fromUri(request.url())
                                .host(target.getHost())
                                .port(target.getPort())
                                .build()
                                .toUri();
                            return next
                                .exchange(ClientRequest
                                              .from(request)
                                              .url(uri)
                                              .build());
                        }
                        return next.exchange(request);
                    }
                })
                .defaultHeaders(header -> {
                    if (StringUtils.hasText(password)) {
                        header.setBasicAuth(username, password);
                    }
                })
                .codecs(clientCodecConfigurer -> clientCodecConfigurer
                    .defaultCodecs()
                    .maxInMemorySize((int) DataSize.ofMegabytes(200).toBytes()))
                .build()
                ;

        }
    }
}
